package cn.itcast.tramsformation;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.tramsformation
 * @date 2021/2/24 0024 10:51
 * @Copyright © 2015-04-29  One for each, and two for each
 *              演示Transformations-合并和连接操作
 */
public class TransformationsDemo02 {
    public static void main(String[] args) throws Exception {
        //TODO 0.ENV
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //TODO 1.soruce
        final DataStreamSource<String> ds1 = env.fromElements("hadoop", "spark", "flink");
        final DataStreamSource<String> ds2 = env.fromElements("hadoop", "spark", "flink");
        final DataStreamSource<Long> ds3 = env.fromElements(1L, 2L, 3L);
        //TODO 2.transformation
        final DataStream<String> result1 = ds1.union(ds2);                  //union能合并同类型，不可以合并不同类型
        final ConnectedStreams<String, String> result2 = ds1.connect(ds2);
        final ConnectedStreams<String, Long> result3 = ds1.connect(ds3);    //connect可以合并同类型，也可以合并不同类型

        final SingleOutputStreamOperator<String> result = result3.map(new CoMapFunction<String, Long, String>() {
            @Override
            public String map1(String value) throws Exception {
                return "String:" + value;
            }

            @Override
            public String map2(Long value) throws Exception {
                return "Long:" + value;
            }
        });
        //TODO 3.SINK
        result1.print();
        //result2.print();//注意:connect之后需要做其他的处理,不能直接输出
        //result3.print();//注意:connect之后需要做其他的处理,不能直接输出
        result.print();
        //TODO 4.execute
        env.execute();
    }
}
